package org.onion.pattern.event.spring.outbox;

import jakarta.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.onion.pattern.event.outbox.config.EventOutboxConfig;
import org.onion.pattern.event.outbox.data.OutboxEventInfo;
import org.onion.pattern.event.outbox.executor.GroupingOutboxDeliveryExecutor;
import org.onion.pattern.event.publish.EventChannelPublisher;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.transaction.support.TransactionTemplate;

import java.util.List;
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;

@Slf4j
@Component
public class JdbcGroupingOutboxDeliveryExecutor extends GroupingOutboxDeliveryExecutor<OutboxEventInfo> {

    @Autowired
    private JdbcOutboxEventStorage jdbcEventStorage;
    @Autowired
    private EventChannelPublisher channelPublisher;
    @Autowired
    private TransactionTemplate transactionTemplate;

    public JdbcGroupingOutboxDeliveryExecutor(EventOutboxConfig config) {
        super(config);
    }

    @PreDestroy
    void cleanup() {
        this.shutdown();
    }

    @Override
    protected ExecStatus runWrapper(ExecStatus execStatus, Function<ExecStatus, ExecStatus> fun) {
        return this.transactionTemplate.execute(transactionStatus -> {
            //                Object savepoint = transactionStatus.createSavepoint();
            try {
                return fun.apply(execStatus);
            } catch (Throwable e) {
                log.error("", e);
                transactionStatus.setRollbackOnly();
                return ExecStatus.INTERVAL;
            }
        });
    }

    @Override
    protected List<OutboxEventInfo> findReadyEvent(String group, int limit) throws Exception {
        //TODO 使用id 偏移 2025-08-02 13:31  何挺强
        return jdbcEventStorage.findReadyEventByGroup(group, limit);
    }

    @Override
    protected void publish(List<OutboxEventInfo> msg) throws Exception {
        if (CollectionUtils.isEmpty(msg)) {
            return;
        }
        //TODO 按通道批量发送 2025-08-18 23:27  何挺强
        for (OutboxEventInfo outBoxEventInfo : msg) {
            this.channelPublisher.publish(outBoxEventInfo);
        }
    }

    @Override
    protected boolean ack(List<OutboxEventInfo> event) {
        Set<String> collect = event.stream().map(OutboxEventInfo::getEventKey).collect(Collectors.toSet());
        this.jdbcEventStorage.ack(collect);
        return true;
    }

}
